package com.atguigu.flink.exec2;

import com.atguigu.flink.pojo.UserBehavior;
import com.atguigu.flink.util.MyUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2022/11/22
 *
     实时统计每小时内的网站PV
            基于EventTime 时间的滚动窗口。窗口的范围(大小)是1h
 */
public class Demo1_PV
{
    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        WatermarkStrategy<UserBehavior> watermarkStrategy = WatermarkStrategy.<UserBehavior>forMonotonousTimestamps()
            .withTimestampAssigner((u, t) -> u.getTimestamp() * 1000);

        env.readTextFile("data/UserBehavior.csv")
           .flatMap(new FlatMapFunction<String, UserBehavior>()
           {
               @Override
               public void flatMap(String value, Collector<UserBehavior> out) throws Exception {
                   String[] words = value.split(",");
                   if ("pv".equals(words[3])){
                       out.collect(new UserBehavior(
                            Long.valueOf(words[0]),
                            Long.valueOf(words[1]),
                            Integer.valueOf(words[2]),
                           words[3],
                            Long.valueOf(words[4])
                       ));
                   }
               }
           })
           .assignTimestampsAndWatermarks(watermarkStrategy)
           .windowAll(TumblingEventTimeWindows.of(Time.hours(1)))
           .aggregate(new AggregateFunction<UserBehavior, Integer, Integer>()
           {
               @Override
               public Integer createAccumulator() {
                   return 0;
               }

               @Override
               public Integer add(UserBehavior value, Integer accumulator) {
                   return accumulator += 1;
               }

               @Override
               public Integer getResult(Integer accumulator) {
                   return accumulator;
               }

               @Override
               public Integer merge(Integer a, Integer b) {
                   return null;
               }
           }, new ProcessAllWindowFunction<Integer, String, TimeWindow>()
           {
               @Override
               public void process(Context context, Iterable<Integer> elements, Collector<String> out) throws Exception {
                   TimeWindow window = context.window();
                   out.collect(MyUtil.parseTimeWindow(window) +":"+elements.iterator().next());

               }
           })
           .print().setParallelism(1);


        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }

    }
}
